#ifndef A8_TCPLISTENER_H
#define A8_TCPLISTENER_H

#include <a8/netbase.hpp>
#include <a8/_socketitem.hpp>
#include <a8/_socketpool.hpp>

namespace a8
{	

	class TcpListener
	{
	public:
		a8::Delegate<void (a8::ClientSocketItem**)> OnCreateClientSocket;
		a8::Delegate<void(const char*, int, bool&)> OnClientConnect;		
		a8::Delegate<void (a8::TcpListener*, int)> OnError;		
		std::string mBindAddress;
		unsigned short mBindPort;
	  
		TcpListener(unsigned short maxClientCnt=0xEFFF):
			mListenSocket(INVALID_SOCKET), 
			mBindAddress(""), 
			mBindPort(0),
			mAcceptThread(NULL), 
			mWorkerThread(NULL), 
			mCurrSocketHandle(1000),
			mClientCount(0), 
			mMaxClients(maxClientCnt)
		{
#ifdef WIN32
			mhIocp = 0;
#endif
#ifdef __linux__
			mhEpoll = 0;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			mhKqueue = 0;
#endif
			INIT_LIST_HEAD(&mClientList);
			INIT_HLIST_BUCKET(mClientsBucket);
		}
		
		virtual ~TcpListener()
		{
			Close();
		}

		int GetClientSocketCount()
		{
			return mClientCount;
		}

		int GetPoolSocketCount()
		{
			return mFreeClientPool.Count();
		}
	 
		void Open()
		{
			if(!IsActive()){
				SetActive(true);
			}
		}
	 
		void Close()
		{
			if(IsActive()){
				SetActive(false);
			}
		}


		void Lock()
		{
			mClientsMutex.Lock();
		}

		void UnLock()
		{
			mClientsMutex.UnLock();
		}
	 
		bool IsActive() 
		{
			return mListenSocket != INVALID_SOCKET;
		}

		void ForceCloseClient(unsigned short sockhandle)
		{
			mClientsMutex.Lock();
			ClientSocketItem *p = GetClientSocket(sockhandle);
			if(p){
				if(p->Socket() != INVALID_SOCKET){
					p->_ForceClose();
				}
			}
			mClientsMutex.UnLock();
		}

		void FreeTimeOutaSocket(std::vector<unsigned short>& activitylist, int seconds)
		{
			mClientsMutex.Lock();
			struct list_head *node;
			list_for_each(node, &mClientList){
				ClientSocketItem *p = list_entry(node, ClientSocketItem, mListEntry);
			}
			mClientsMutex.UnLock();
		}
		
		bool SendClientMsg(unsigned short sockhandle, const char *buff, int buffLen)
		{
			bool boSendOk = false;
			mClientsMutex.Lock(); 
			ClientSocketItem *p = GetClientSocket(sockhandle);
			if (p){
				p->SendBuff(buff, buffLen);
				boSendOk = true;
			}		 
			mClientsMutex.UnLock();
			return boSendOk;
		}

		void MarkClientIsActivate(unsigned short sockhandle)
		{
			mClientsMutex.Lock(); 
			ClientSocketItem *p = GetClientSocket(sockhandle);
			if (p){
				p->mIsActivite = true;
			}		 
			mClientsMutex.UnLock();
		}

		bool SendClientMsg(unsigned short sockhandle, const std::string& buff)
		{
			return SendClientMsg(sockhandle, buff.data(), buff.size());
		}

		void BroadcastMsg(const char* buff, int bufflen)
		{
			mClientsMutex.Lock(); 
			struct list_head *node = NULL;
			list_for_each(node, &mClientList){
				ClientSocketItem *p = list_entry(node, ClientSocketItem, mListEntry);
				p->SendBuff(buff, bufflen);
			}
			mClientsMutex.UnLock();
		}

		void ForeachSocket(a8::Delegate<void (a8::ClientSocketItem*, std::vector<unsigned short>&) >& func,
                           std::vector<unsigned short>& vec)
		{
			Lock();
			struct list_head *node;
			list_for_each(node, &mClientList){
				ClientSocketItem *p = list_entry(node, ClientSocketItem, mListEntry);
				func(p, vec);
			}
			UnLock();
		}

	private:
		SOCKET mListenSocket;
		a8::Runnable* mAcceptThread;
		a8::Runnable* mWorkerThread;
		a8::sync::Mutex mClientsMutex;	
		hlist_head mClientsBucket[1024];
		std::map<ClientSocketItem*, ClientSocketItem*> mClientHash;
		list_head mClientList;
		volatile unsigned int mClientCount;
		unsigned short mCurrSocketHandle;
		unsigned short mMaxClients;
		ClientSocketPool mFreeClientPool;
		ClientSocketItem* mDeadSocketPool[1024];
#ifdef __linux__
		int mhEpoll;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
		int mhKqueue;
#endif
#ifdef WIN32
		HANDLE mhIocp;
#endif

		static void FreeClientWithNoLock(void* ownerlistener, ClientSocketItem *p)
		{
			((TcpListener*)ownerlistener)->FreeClientWithNoLock(p);
		}

		static void LockClients(void* ownerlistener)
		{
			((TcpListener*)ownerlistener)->Lock();
		}

		static void UnLockClients(void* ownerlistener)
		{
			((TcpListener*)ownerlistener)->UnLock();
		}

		void FreeClientWithNoLock(ClientSocketItem *p)
		{
			std::map<ClientSocketItem*, ClientSocketItem*>::iterator itr = mClientHash.find(p);
			if(itr != mClientHash.end()){
				mClientHash.erase(itr);
				if(p->Socket() != INVALID_SOCKET){
					p->_ForceClose();
				}
				p->Finalize();
				p->OnDisConnect();
				list_del(&p->mListEntry);
				hlist_del(&p->mHlistEntry);
				mFreeClientPool.Add(p);
				mClientCount--;
			}
		}

		//支持重入多次调用
		void FreeClient(ClientSocketItem *p)
		{
			mClientsMutex.Lock();
			FreeClientWithNoLock(p);
			mClientsMutex.UnLock();
		}
	  
		void SetActive(bool active) 
		{ 
			if(active){
				if(IsActive()){
					return;
				}
				if(!mWorkerThread){
					mWorkerThread = new a8::Runnable();
					mWorkerThread->OnExecute += std::make_pair(this, &TcpListener::WorkerThreadProc);
					mWorkerThread->Start();
				}
			}else{
				if(IsActive()){
					ActiveStop();		
				}
			}
		}

		bool ActiveStart()
		{
#ifdef WIN32
			mListenSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
#else
			mListenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
#endif
			if(INVALID_SOCKET == mListenSocket){
				OnError(this, errno);
				return false;
			}
			// TIME_WAIT - argh
			int on = 1;
			if (setsockopt(mListenSocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on)) == -1 ){
				//return false;
			}
			sockaddr_in sa;
			memset(&sa, 0, sizeof(sa));
			sa.sin_family = AF_INET;
			sa.sin_addr.s_addr = inet_addr(mBindAddress.c_str());
			sa.sin_port = htons(mBindPort);
			if(bind(mListenSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
				OnError(this, errno);
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;
				return false;
			}
			if(listen(mListenSocket, mMaxClients) < 0) {
				OnError(this, errno);
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;
				return false;
			}
#ifdef __linux__
			mhEpoll = epoll_create(mMaxClients);
			assert(mhEpoll != -1);
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			mhKqueue = kqueue();
#endif
#ifdef WIN32
			mhIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);
			assert(mhIocp);
#endif		
			mAcceptThread = new a8::Runnable();
			mAcceptThread->OnExecute += std::make_pair(this, &TcpListener::AcceptThreadProc);
			mAcceptThread->Start();
			return true;
		}

		void ActiveStop()
		{
#ifdef WIN32
			if (mhIocp){
				PostQueuedCompletionStatus(mhIocp, 0, NULL, NULL);
			}
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			if(mhKqueue){
				close(mhKqueue);
				mhKqueue = 0;
			}
#endif
			if(mListenSocket != INVALID_SOCKET){
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;								
			}
			if (mAcceptThread){
				delete mAcceptThread;
				mAcceptThread = NULL;
			}		
			if (mWorkerThread){
				delete mWorkerThread;
				mWorkerThread = NULL;
			}
#ifdef __linux__
			if(mhEpoll != -1){
				close(mhEpoll);
				mhEpoll = -1;
			}
#endif
#ifdef WIN32
			if (mhIocp){
				CloseHandle(mhIocp);
				mhIocp = NULL;
			}
#endif		
		}
	
		void AcceptThreadProc(a8::Runnable* thread)
		{
			sockaddr_in addr;
			socklen_t addrLen = sizeof(sockaddr_in);
			while(!thread->IsShutdown()){
				if(INVALID_SOCKET == mListenSocket){
					break;	
				}
				addrLen = sizeof(sockaddr_in);
				memset(&addr, 0, sizeof(addr));
				SOCKET sock = accept(mListenSocket, (sockaddr*)&addr, (socklen_t*)&addrLen);
				if(INVALID_SOCKET != sock){
					if (thread->IsShutdown()){
						closesocket(sock);
					}else{
						if(!SocketAccept(sock, inet_ntoa(addr.sin_addr), addr.sin_port, addr.sin_addr.s_addr)){
							closesocket(sock);
						}
					}
				}
			}
		}

		void WorkerThreadProc(a8::Runnable* thread)
		{
			if(!ActiveStart()){
				return;
			}
#ifdef WIN32
			while (!thread->IsShutdown()){
				DWORD completeBytes = 0;
				OVERLAPPED *ol = NULL;
				ClientSocketItem *clientsocket = NULL;
				BOOL boSuccess = GetQueuedCompletionStatus(mhIocp, &completeBytes,(PULONG_PTR)&clientsocket,(LPOVERLAPPED*)&ol, INFINITE);
				if(ol == (LPOVERLAPPED)DISCONNECT_FLAG && clientsocket){
					FreeClient(clientsocket);
					continue;		
				}
				// I/O¶ÁÐ´·µ»Ø0£¬¿ÉÄÜ³¬Ê±£¬¿ÉÄÜ¿Í»§¶Ë¶Ï¿ªÁ¬½Ó¡£  µÚÒ»¸ö·´Ó¦
				if(!boSuccess || completeBytes == 0){
					if(clientsocket){
						clientsocket->_ForceClose();
					}
					continue;
				}
				if (clientsocket && ol){
					IocpOverLapped* iocpOl = container_of(ol, IocpOverLapped, mOverLapped);
					clientsocket->mActiveTick = 0;
					switch(iocpOl->mOpCode){
					case IOCP_RECV:		
						{
							clientsocket->DoClientRecv(completeBytes);
						}
						break;
					case IOCP_SEND:
						{
							clientsocket->DoClientSend(completeBytes);
						}
						break;
					}
				}
			}	 	  
#endif
#ifdef __linux__
			epoll_event *events = new epoll_event[mMaxClients];
			while(!thread->IsShutdown()){
				int nfds = epoll_wait(mhEpoll, events, mMaxClients, -1);
				for(int i = 0; i < nfds; i++){	  
					ClientSocketItem *clientsocket = (ClientSocketItem*)events[i].data.ptr;
					if(events[i].events & EPOLLIN){
						clientsocket->DoClientRecv();
					}else if(events[i].events & EPOLLOUT){
						clientsocket->DoClientSend();
					}else if(events[i].events & EPOLLRDHUP || events[i].events & EPOLLHUP){
						FreeClient(clientsocket);
					}
				}
			}
			delete [] events;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			struct kevent *events = (struct kevent*)malloc(sizeof(struct kevent) * mMaxClients);
			while(!thread->IsShutdown()){
				int nEvents = kevent(mhKqueue, NULL, 0, events, mMaxClients, NULL);
				if(nEvents == -1)
					break;
				for(int i = 0; i < nEvents; i++){
				}
			}
			free(events);
#endif
		}

		bool SocketAccept(SOCKET sock, const char* ipAddr, int port, unsigned long saddr)
		{		
			if(mClientCount >= mMaxClients){
				return false;
			}
			bool refuse = false;
			OnClientConnect(ipAddr, port, refuse);
			if (refuse){
				return false;
			}

			time_t nowtime = time(NULL);
#if 1
			mClientsMutex.Lock();
			{
				int deadCnt = 0;
				struct list_head *node;
				list_for_each(node, &mClientList){
					ClientSocketItem *p = list_entry(node, ClientSocketItem, mListEntry);
					if(p && !p->mIsActivite && nowtime - p->mCreateTime > SOCKET_TIMEOUT){
						if(deadCnt < 1024){
							mDeadSocketPool[deadCnt++] = p;
						}
					}
				}
				while(deadCnt > 0){
					mDeadSocketPool[deadCnt - 1]->_ForceClose();
					deadCnt--;
				}
			}
			mFreeClientPool.ClearTimeOutSocket();
			ClientSocketItem* p = mFreeClientPool.Get();
			mClientsMutex.UnLock();

			if(!p){
				OnCreateClientSocket(&p);
			}
			if (!p){
				return false;
			}
#else			
			ClientSocketItem* p = NULL;
			OnCreateClientSocket(&p);
			if(!p){
				return false;
			}
#endif
			p->mCreateTime = nowtime;
			p->mOwnerListener = this;
			p->mLockClients = &TcpListener::LockClients;
			p->mUnLockClients = &TcpListener::UnLockClients;
			p->mFreeClientWithNoLock = &TcpListener::FreeClientWithNoLock;
			p->Initialize();			
			p->SetSocket(sock);
			p->mSAddr = saddr;
			p->mRemoteAddress = std::string(ipAddr);
			p->mRemotePort = port;

			mClientsMutex.Lock();
			++mCurrSocketHandle;			
			while (GetClientSocket(mCurrSocketHandle)){
				if (mCurrSocketHandle < 1000){
					mCurrSocketHandle = 1000;
				}
				++mCurrSocketHandle;			
			}	
			p->mSocketHandle = mCurrSocketHandle;
			list_add(&p->mListEntry, &mClientList);		
			int h = p->mSocketHandle % a8::arraysize(mClientsBucket);			
			hlist_add_head(&p->mHlistEntry, &mClientsBucket[h]);
			mClientCount++;						
			mClientHash.insert(std::make_pair(p, p));
			assert(GetClientSocket(p->mSocketHandle));
			mClientsMutex.UnLock();
			p->OnConnect();
#ifdef WIN32
			p->mhIocp = mhIocp;
			if(CreateIoCompletionPort((HANDLE)sock, mhIocp, (ULONG_PTR)p, 0) == 0){
				p->_ForceClose();
			}else{
				p->PostRecv();
			}
#endif
#ifdef __linux__
			int flags = 0;
			flags = fcntl(sock, F_GETFL, 0);
			fcntl(sock, flags|O_NONBLOCK);
			p->mhEpoll = mhEpoll;
			struct epoll_event ev;
			ev.data.fd = sock;
			ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
			ev.data.ptr = p;
			int n = epoll_ctl(mhEpoll, EPOLL_CTL_ADD, sock, &ev);
			assert(n == 0);
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			p->mhKqueue = mhKqueue;
			struct kevent kvt;
			memset(&kvt, 0, sizeof(struct kevent));
			EV_SET(&kvt, sock, EVFILT_READ, EV_ADD | EV_ENABLE | EV_ERROR, 0, 0, p);
			kevent(mhKqueue, &kvt, 1, NULL, 0, NULL);
			memset(&kvt, 0, sizeof(struct kevent));
			EV_SET(&kvt, sock, EVFILT_WRITE, EV_ADD | EV_ENABLE | EV_ERROR, 0, 0, p);
			kevent(mhKqueue, &kvt, 1, NULL, 0, NULL);
#endif
			return true;
		}
	  
		ClientSocketItem* GetClientSocket(unsigned short handle)
		{
			int h = handle % a8::arraysize(mClientsBucket);
			hlist_head *hlist = &mClientsBucket[h];
			hlist_node *node = NULL;
			hlist_for_each(node, hlist){
				ClientSocketItem* p = hlist_entry(node, ClientSocketItem, mHlistEntry);
				if (p->mSocketHandle == handle)
					return p;			
			}
			return NULL;
		}
		
	};

}

#endif
